package com.zhengbo.flink.demo.mq.cardata;

import com.alibaba.fastjson.JSONObject;
import com.zhengbo.flink.demo.base.utils.GpsCorrect;
import com.zhengbo.flink.demo.mq.bean.CarStatus;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.concurrent.ThreadLocalRandom;

/**
 * 处理车辆消息数据
 * 车辆消息映射为CarStatus对象
 * 类似于jstorm中spout
 * @author zhengbo
 * @date 2019/8/15
 */
@Slf4j
public class CarDataMapFunction extends RichFlatMapFunction<String, CarStatus> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        log.info("CarDataMapFunction open success....");
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void flatMap(String value, Collector<CarStatus> out) throws Exception {

        log.info("flatMap receive message:{}", value);

        CarStatus carStatus = JSONObject.parseObject(value, CarStatus.class);

        ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
        int randomDriverNo = threadLocalRandom.nextInt();

        double[] latlon = new double[2];
        GpsCorrect.transform(carStatus.getLat(), carStatus.getLon(), latlon);
        carStatus.setLat(latlon[0]);
        carStatus.setLon(latlon[1]);

        carStatus.setDriverNo(String.valueOf(randomDriverNo));

        log.info("collect car status data:{}", JSONObject.toJSONString(carStatus));

        out.collect(carStatus);
    }
}
